package com.king.first.app.windows;

import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * 增量聚合函数
 */
public class Window_Function_Aggregate {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> socketTextStream = env.socketTextStream("hadoop01", 9999);
        SingleOutputStreamOperator<Tuple2<String, Integer>> flatMapStream = socketTextStream.flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (v, collector) -> {
            String[] s = v.split(" ");
            for (String s1 : s) {
                collector.collect(Tuple2.of(s1, 1));
            }
        }).returns(Types.TUPLE(Types.STRING, Types.INT));
        KeyedStream<Tuple2<String, Integer>, String> keyedStream = flatMapStream.keyBy(v -> v.f0);
        SingleOutputStreamOperator<Tuple2<String, Integer>> reduce = keyedStream.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .aggregate(new AggregateFunction<Tuple2<String, Integer>, Integer, Tuple2<String, Integer>>() {
                    private String key = "";
                    @Override
                    public Integer createAccumulator() {
                        return 0;
                    }
                    @Override
                    public Integer add(Tuple2<String, Integer> tuple2, Integer integer) {
                        key = tuple2.f0;
                        return tuple2.f1 + integer;
                    }
                    @Override
                    public Tuple2<String, Integer> getResult(Integer integer) {
                        return Tuple2.of(key, integer);
                    }
                    @Override
                    public Integer merge(Integer integer, Integer acc1) {
                        return integer + acc1;
                    }
                });
        reduce.print();
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
